package database.team.phase2;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;


public class SortedJoinBlocks {
	static int counterIOEmpl = 0;
    static int counterIODept = 0;
    static int totalIOJoin = 0;
	public int SortedJoin() throws IOException{
		String valuePK="",tupleContent,valueFK = "";
		int empIDFKDept=0,empIDPK = 0;
		long sizeFileDept;
		long sizeFileEmp;
		File fileDept = new File("FileSortedAndMerged_TableDept.txt");
		File fileEmp = new File("FileSortedAndMerged_TableEmp.txt");
		sizeFileEmp = fileEmp.length();
		sizeFileDept = fileDept.length();
		FileWriter output = new FileWriter("SortedJoinBlocks.txt");
		PrintWriter tuple = new PrintWriter(output);
		
		long sizeoffileEmp = fileEmp.length();
        long blocksizeEmp = sizeoffileEmp / 1024 + (sizeoffileEmp % 1024 == 0 ? 0 : 1);
        long freemem = Runtime.getRuntime().freeMemory();
        if (blocksizeEmp < freemem / 2) {
                blocksizeEmp = freemem/4;
        }
        
        long sizeoffileDept = fileDept.length();
        long blocksizeDept = sizeoffileDept / 1024 + (sizeoffileDept % 1024 == 0 ? 0 : 1);
        long freemem1 = Runtime.getRuntime().freeMemory();
        if (blocksizeDept < freemem1 / 2) {
                blocksizeDept = freemem1/4;
        }
   		
		//Perform a join taking table Department as a Relation R and Employee as a relation S
		if(sizeFileDept <= sizeFileEmp){
			BufferedReader Dept = new BufferedReader(new FileReader(fileDept));
			valueFK = Dept.readLine();
			
		    while ((valueFK != null)){
				long currentblocksizeDept = 0;
				BufferedReader Emp = new BufferedReader(new FileReader(fileEmp));
				valuePK = Emp.readLine();
				while ((valueFK !=null) && (currentblocksizeDept < blocksizeDept)){
					currentblocksizeDept += valueFK.length() * 2;
					empIDFKDept = Integer.parseInt(valueFK.substring(0, 7));
					if(valuePK !=null){
						empIDPK = Integer.parseInt(valuePK.substring(0, 7));
					}
					while ((valuePK != null) &&(valueFK!=null) && (empIDPK <= empIDFKDept)){
						long currentblocksizeEmp = 0;
						counterIODept++;
						while ((valuePK != null) &&(valueFK!=null)&& (currentblocksizeEmp < blocksizeEmp)&& (empIDPK <= empIDFKDept)){
								empIDPK = Integer.parseInt(valuePK.substring(0, 7));
								if(empIDPK == empIDFKDept){
									tupleContent =valuePK+valueFK;	
									tuple.println(tupleContent);
									valueFK = Dept.readLine();
									if(valueFK !=null){
										   empIDFKDept = Integer.parseInt(valueFK.substring(0, 7));
									}

								}
								else{
									valuePK = Emp.readLine();
									if(valuePK !=null){
									   empIDPK = Integer.parseInt(valuePK.substring(0, 7));
									   currentblocksizeEmp += valuePK.length() * 2;
									}
								}
								
						}

					}
					valueFK = Dept.readLine();
						
				}
					
			}
		    
		  //Closing tuple and output readers and writers
		  tuple.close(); output.close();
		  return totalIOJoin = counterIODept;
		   
		}
		
	
	//Perform a join taking table Employee as a Relation R and Department as a relation S
		else{
			
			BufferedReader Emp = new BufferedReader(new FileReader(fileEmp));
		    valuePK = Emp.readLine();
		    BufferedReader Dept = new BufferedReader(new FileReader(fileDept));
			valueFK = Dept.readLine();
			    while ((valuePK != null)){
					long currentblocksizeEmp = 0;
					counterIOEmpl++;
					while ((valuePK !=null) && (valueFK != null)&& (currentblocksizeEmp < blocksizeEmp)){
						empIDPK = Integer.parseInt(valuePK.substring(0, 7));
						currentblocksizeEmp += valuePK.length() * 2;
						if(valuePK !=null){
							empIDFKDept = Integer.parseInt(valueFK.substring(0, 7));
						}
						while ((valueFK != null)&& (valuePK !=null) && (empIDPK >= empIDFKDept)){
							long currentblocksizeDept = 0;
							//counterIODept++;
							while ((valueFK != null) && (valuePK !=null) &&(currentblocksizeDept < blocksizeDept)&& (empIDPK >= empIDFKDept)){
									empIDFKDept = Integer.parseInt(valueFK.substring(0, 7));
									if(empIDPK == empIDFKDept){
										tupleContent =valuePK+valueFK;	
										tuple.println(tupleContent);
										valueFK = Dept.readLine();
										if(valueFK != null){
											empIDFKDept = Integer.parseInt(valueFK.substring(0, 7));
											currentblocksizeDept += valueFK.length() * 2;
										}
									}
									
							}
						}
						valuePK = Emp.readLine();
						
					}
					
				}
			    
			}
			  //Closing tuple and output readers and writers
			  tuple.close(); output.close();
			  return totalIOJoin = counterIOEmpl;
		}	
}
			
		/*	BufferedReader Emp = new BufferedReader(new FileReader(fileEmp));
			    valuePK = Emp.readLine();
			    while ((valuePK != null)){
					long currentblocksizeEmp = 0;
					BufferedReader Dept = new BufferedReader(new FileReader(fileDept));
					valueFK = Dept.readLine();
					counterIOEmpl++;
					while ((valuePK !=null) && (currentblocksizeEmp < blocksizeEmp)){
						empIDPK = Integer.parseInt(valuePK.substring(0, 7));
						currentblocksizeEmp += valuePK.length() * 2;
						if(valuePK !=null){
							empIDFKDept = Integer.parseInt(valueFK.substring(0, 7));
						}
						while ((valueFK != null)&& (empIDPK >= empIDFKDept)){
							long currentblocksizeDept = 0;
							counterIODept++;
							while ((valueFK != null) && (currentblocksizeDept < blocksizeDept)&& (empIDPK >= empIDFKDept)){
									empIDFKDept = Integer.parseInt(valueFK.substring(0, 7));
									if(empIDPK == empIDFKDept){
										tupleContent =valuePK+valueFK;	
										tuple.println(tupleContent);
									}
									valueFK = Dept.readLine();
									if(valueFK != null){
										empIDFKDept = Integer.parseInt(valueFK.substring(0, 7));
										currentblocksizeDept += valueFK.length() * 2;
									}
									
							}
	
						}
						valuePK = Emp.readLine();
						
					}
					
				}
			    
			}
			  //Closing tuple and output readers and writers
			  tuple.close(); output.close();
			  return totalIOJoin = counterIODept+counterIOEmpl;
		
		}  
			
		public static void main(String[] args) throws IOException {
			 System.out.println("Joining the table employee with Department");
			 long startTime = System.currentTimeMillis();
			 SortedJoinBlocks sjb = new SortedJoinBlocks();
			 sjb.SortedJoin();
			 long endTime = System.currentTimeMillis();
			 float totalTime = (endTime - startTime);
			 System.out.println("Total Disk I/O to Join the tables: "+(totalIOJoin)); 
			 System.out.println("Total time to Join the tables: "+totalTime); 
			 System.out.println("The result is shown in the file SortedJoinBlocks.txt");
		}*/
	

		
		

